package com.avic.szzx.service;

import com.alibaba.fastjson.JSONObject;
import com.avic.szzx.util.EsUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.LinkedHashMap;
import java.util.List;

@Slf4j
@Service
public class KafkaService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 通过Kafka生产者发送消息到指定topic
     * @param status
     * @param type
     * @param data
     * @param tag
     */
    public void sendKafkaMessage(Boolean status,String type,Object data,String tag,String option) {
        LinkedHashMap<String, Object> stringObjectMap = EsUtils.ObjToMap(data);
        stringObjectMap.put("tag", tag);
        stringObjectMap.put("option",option);
        String JsonString = JSONObject.toJSONString(stringObjectMap);
        if (status) {
            if (type.equals("VERTEX")) {
                sendMessageAsync("test_vertex", JsonString);
            } else {
                sendMessageAsync("test_edge", JsonString);
            }
        }
    }
    public void sendKafkaMessages(Boolean status, String type, List<String> datas, String tag) {
        if (status) {
            if (type.equals("VERTEX")) {
                sendMessageAsyncs("test_vertexs", datas);
            }
        }
    }
    /**
     * producer 异步方式发送数据
     * @param topic   topic名称
     * @param message producer发送的数据
     */
    public void sendMessageAsync(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
    public void sendMessageAsyncs(String topic, List<String> messages) {
        kafkaTemplate.send(topic, messages);
    }
}
